package filter;

import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;

import java.util.Collections;

/**
 * 2 * @Author: 王杰
 * 3 * @Date: 2020/11/18 13:19
 * 4
 */
public class MyFilter extends RichFilterFunction<String> implements CheckpointedFunction {
    //定义生成数据的标记
    boolean isRunning = true;

    public MyFilter() {
        super();
    }

    //定义一个state，里面放的就是offset，将偏移量保存到hdfs上
    ListState<Long> offsetState = null;
    //定义偏移量
    Long offset = 0L;

    @Override
    public boolean filter(String value) throws Exception {
        Iterable<Long> longs = offsetState.get();
        boolean b = offsetState.get().equals(Collections.singletonList(1L));
        if (value.contains("SampleTime") && !b) {
            offsetState.update(Collections.singletonList(1L));
            return true;
        } else if (value.contains("SampleTime") && b) {
            return false;
        }
        return true;
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        offsetState.clear();
        offsetState.add(offset);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        //初始化状态或恢复历史状态
        //定义一个状态描述器
        ListStateDescriptor<Long> descriptor = new ListStateDescriptor<Long>(
                "offsetState",
                TypeInformation.of(new TypeHint<Long>() {
                })
        );
        offsetState = context.getOperatorStateStore().getListState(descriptor);
    }
}
